#!/usr/bin/python2.7
# -*- coding: utf-8 -*
#-------------------------------------------------------
# 程序：KafkaProducer
# 作者：郭凯凯
# 文件：kafka_producer.py
# 日期：2018/2/22
# 版本：1.0.0
# 环境：python 2.7
# 组件：kafka
# 描述：实现KafkaProducer
# 备注：
#-------------------------------------------------------
#
# producer = KafkaProducer(bootstrap_servers = ['127.0.0.1:9092'])
# topic = 'psaems.power.useinfo'

import sys
import json
from kafka import KafkaProducer


class KProducer():
    def __init__(self):
        self.servers = None
        self.topic = None
        self.producer = None

    def __del__(self):
        self.Destroty()

    def Destroty(self):
        print 'Destroty'
        #self.producer.close()

    def Connect(self, b_servers, d_topic, s_p_username=None,
                s_p_password=None):
        """
        @summary:连接kafka.
        @param bootstrap_servers: IP地址.
        @param topic: 主题
        @param sasl_plain_username = psaems
        @param sasl_plain_password = psaems!^*
        @return: 连接成功返回1，否则返回-1.
        """
        try:
            self.topic = d_topic 
            if s_p_username is None or s_p_password is None:
                self.producer = KafkaProducer(bootstrap_servers=b_servers)
            else:
                self.producer = KafkaProducer(
                    bootstrap_servers=b_servers,
                    sasl_plain_username=s_p_username,
                    sasl_plain_password=s_p_password,
                    api_version=(0,10,2),
                    security_protocol='PASL_PLAINTEXT',
                    sasl_mechanism='PLAIN'
                    )
                print 'success'
            return self.producer
        except Exception as e:
            return e

    def sendInfo(self, info):
        """
        @summary:发送信息给kafka.
        @param info: 信息.
        @return: 连接成功返回1，否则返回-1.
        """
        try:
            reslut = self.producer.send(self.topic, info)
            self.producer.flush()
            print ('*****************'+reslut)
            return reslut
        except Exception as e:
            return e

    def sendjsondata(self, params):
        """
        @summary:发送Json信息给kafka.
        @param info: 信息.
        @return: 连接成功返回1，否则返回-1.
        """
        try:
            parmas_message = json.dumps(params)
            producer = self.producer
            reslut = producer.send(self.topic, parmas_message.encode('utf-8'))
            print ('*****************'+reslut)
            producer.flush()
            return reslut
        except Exception as e:
            return e
